/*-------------------------------------------------------------------------
 *
 * spgxlog.c
 *	  WAL replay logic for SP-GiST
 *
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *			 src/backend/access/spgist/spgxlog.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "access/bufmask.h"
#include "access/spgist_private.h"
#include "access/spgxlog.h"
#include "access/transam.h"
#include "access/xlog.h"
#include "access/xlogutils.h"
#include "storage/standby.h"
#include "utils/memutils.h"


static MemoryContext opCtx;		/* 工作内存用于操作 */


/*
 * 准备一个虚拟的 SpGistState，包含重放所需的最低信息。
 *
 * 目前，我们只需要足够的信息来支持 spgFormDeadTuple()，
 * 以及 isBuild 标志。
 */
static void fc_fillFakeState(SpGistState *fc_state, spgxlogState fc_stateSrc)
{
	memset(fc_state, 0, sizeof(*fc_state));

	fc_state->myXid = fc_stateSrc.myXid;
	fc_state->isBuild = fc_stateSrc.isBuild;
	fc_state->deadTupleStorage = palloc0(SGDTSIZE);
}

/*
 * 添加一个叶子元组，或替换现有的占位元组。这用于重放 SpGistPageAddNewItem() 操作。
 * 如果偏移量指向一个现有的元组，那么最好是一个占位元组。
 */
static void fc_addOrReplaceTuple(Page fc_page, Item fc_tuple, int fc_size, OffsetNumber fc_offset)
{
	if (fc_offset <= PageGetMaxOffsetNumber(fc_page))
	{
		SpGistDeadTuple fc_dt = (SpGistDeadTuple) PageGetItem(fc_page,
														   PageGetItemId(fc_page, fc_offset));

		if (fc_dt->tupstate != SPGIST_PLACEHOLDER)
			elog(ERROR, "SPGiST tuple to be replaced is not a placeholder");

		Assert(SpGistPageGetOpaque(fc_page)->nPlaceholder > 0);
		SpGistPageGetOpaque(fc_page)->nPlaceholder--;

		PageIndexTupleDelete(fc_page, fc_offset);
	}

	Assert(fc_offset <= PageGetMaxOffsetNumber(fc_page) + 1);

	if (PageAddItem(fc_page, fc_tuple, fc_size, fc_offset, false, false) != fc_offset)
		elog(ERROR, "failed to add item of size %u to SPGiST index page",
			 fc_size);
}

static void fc_spgRedoAddLeaf(XLogReaderState *fc_record)
{
	XLogRecPtr	fc_lsn = fc_record->EndRecPtr;
	char	   *fc_ptr = XLogRecGetData(fc_record);
	spgxlogAddLeaf *fc_xldata = (spgxlogAddLeaf *) fc_ptr;
	char	   *fc_leafTuple;
	SpGistLeafTupleData fc_leafTupleHdr;
	Buffer		fc_buffer;
	Page		fc_page;
	XLogRedoAction fc_action;

	fc_ptr += sizeof(spgxlogAddLeaf);
	fc_leafTuple = fc_ptr;
	/* 叶子元组未对齐，因此需要复制以访问其头部 */
	memcpy(&fc_leafTupleHdr, fc_leafTuple, sizeof(SpGistLeafTupleData));

	/*
	 * 在正常操作中，我们会同时锁定当前页和父页；但是在 WAL 重放中，在更新父页之前更新叶子页应该是安全的。
	 */
	if (fc_xldata->newPage)
	{
		fc_buffer = XLogInitBufferForRedo(fc_record, 0);
		SpGistInitBuffer(fc_buffer,
						 SPGIST_LEAF | (fc_xldata->storesNulls ? SPGIST_NULLS : 0));
		fc_action = BLK_NEEDS_REDO;
	}
	else
		fc_action = XLogReadBufferForRedo(fc_record, 0, &fc_buffer);

	if (fc_action == BLK_NEEDS_REDO)
	{
		fc_page = BufferGetPage(fc_buffer);

		/* 插入新元组 */
		if (fc_xldata->offnumLeaf != fc_xldata->offnumHeadLeaf)
		{
			/* 正常情况下，由 SpGistPageAddNewItem 添加元组 */
			fc_addOrReplaceTuple(fc_page, (Item) fc_leafTuple, fc_leafTupleHdr.size,
							  fc_xldata->offnumLeaf);

			/* 如有需要，更新头元组的链链接 */
			if (fc_xldata->offnumHeadLeaf != InvalidOffsetNumber)
			{
				SpGistLeafTuple fc_head;

				fc_head = (SpGistLeafTuple) PageGetItem(fc_page,
													 PageGetItemId(fc_page, fc_xldata->offnumHeadLeaf));
				Assert(SGLT_GET_NEXTOFFSET(fc_head) == SGLT_GET_NEXTOFFSET(&fc_leafTupleHdr));
				SGLT_SET_NEXTOFFSET(fc_head, fc_xldata->offnumLeaf);
			}
		}
		else
		{
			/* 替换一个 DEAD 元组 */
			PageIndexTupleDelete(fc_page, fc_xldata->offnumLeaf);
			if (PageAddItem(fc_page,
							(Item) fc_leafTuple, fc_leafTupleHdr.size,
							fc_xldata->offnumLeaf, false, false) != fc_xldata->offnumLeaf)
				elog(ERROR, "failed to add item of size %u to SPGiST index page",
					 fc_leafTupleHdr.size);
		}

		PageSetLSN(fc_page, fc_lsn);
		MarkBufferDirty(fc_buffer);
	}
	if (BufferIsValid(fc_buffer))
		UnlockReleaseBuffer(fc_buffer);

	/* 如有必要，更新父级下行链接 */
	if (fc_xldata->offnumParent != InvalidOffsetNumber)
	{
		if (XLogReadBufferForRedo(fc_record, 1, &fc_buffer) == BLK_NEEDS_REDO)
		{
			SpGistInnerTuple fc_tuple;
			BlockNumber fc_blknoLeaf;

			XLogRecGetBlockTag(fc_record, 0, NULL, NULL, &fc_blknoLeaf);

			fc_page = BufferGetPage(fc_buffer);

			fc_tuple = (SpGistInnerTuple) PageGetItem(fc_page,
												   PageGetItemId(fc_page, fc_xldata->offnumParent));

			spgUpdateNodeLink(fc_tuple, fc_xldata->nodeI,
							  fc_blknoLeaf, fc_xldata->offnumLeaf);

			PageSetLSN(fc_page, fc_lsn);
			MarkBufferDirty(fc_buffer);
		}
		if (BufferIsValid(fc_buffer))
			UnlockReleaseBuffer(fc_buffer);
	}
}

static void fc_spgRedoMoveLeafs(XLogReaderState *fc_record)
{
	XLogRecPtr	fc_lsn = fc_record->EndRecPtr;
	char	   *fc_ptr = XLogRecGetData(fc_record);
	spgxlogMoveLeafs *fc_xldata = (spgxlogMoveLeafs *) fc_ptr;
	SpGistState fc_state;
	OffsetNumber *fc_toDelete;
	OffsetNumber *fc_toInsert;
	int			fc_nInsert;
	Buffer		fc_buffer;
	Page		fc_page;
	XLogRedoAction fc_action;
	BlockNumber fc_blknoDst;

	XLogRecGetBlockTag(fc_record, 1, NULL, NULL, &fc_blknoDst);

	fc_fillFakeState(&fc_state, fc_xldata->stateSrc);

	fc_nInsert = fc_xldata->replaceDead ? 1 : fc_xldata->nMoves + 1;

	fc_ptr += SizeOfSpgxlogMoveLeafs;
	fc_toDelete = (OffsetNumber *) fc_ptr;
	fc_ptr += sizeof(OffsetNumber) * fc_xldata->nMoves;
	fc_toInsert = (OffsetNumber *) fc_ptr;
	fc_ptr += sizeof(OffsetNumber) * fc_nInsert;

	/* 现在指针指向叶子元组的列表 */

	/*
	 * 在正常操作中，我们会同时锁定所有三个页面（源、目标和父页面）；但是在 WAL 重放中，只要按正确顺序更新它们，逐个更新应该是安全的。
	 */

	/* 在目标页面插入元组（优先执行，以确保重定向有效） */
	if (fc_xldata->newPage)
	{
		fc_buffer = XLogInitBufferForRedo(fc_record, 1);
		SpGistInitBuffer(fc_buffer,
						 SPGIST_LEAF | (fc_xldata->storesNulls ? SPGIST_NULLS : 0));
		fc_action = BLK_NEEDS_REDO;
	}
	else
		fc_action = XLogReadBufferForRedo(fc_record, 1, &fc_buffer);

	if (fc_action == BLK_NEEDS_REDO)
	{
		int			fc_i;

		fc_page = BufferGetPage(fc_buffer);

		for (fc_i = 0; fc_i < fc_nInsert; fc_i++)
		{
			char	   *fc_leafTuple;
			SpGistLeafTupleData fc_leafTupleHdr;

			/*
			 * 元组未对齐，因此必须复制以访问大小字段。
			 */
			fc_leafTuple = fc_ptr;
			memcpy(&fc_leafTupleHdr, fc_leafTuple,
				   sizeof(SpGistLeafTupleData));

			fc_addOrReplaceTuple(fc_page, (Item) fc_leafTuple,
							  fc_leafTupleHdr.size, fc_toInsert[fc_i]);
			fc_ptr += fc_leafTupleHdr.size;
		}

		PageSetLSN(fc_page, fc_lsn);
		MarkBufferDirty(fc_buffer);
	}
	if (BufferIsValid(fc_buffer))
		UnlockReleaseBuffer(fc_buffer);

	/* 从源页面删除元组，插入重定向指针 */
	if (XLogReadBufferForRedo(fc_record, 0, &fc_buffer) == BLK_NEEDS_REDO)
	{
		fc_page = BufferGetPage(fc_buffer);

		spgPageIndexMultiDelete(&fc_state, fc_page, fc_toDelete, fc_xldata->nMoves,
								fc_state.isBuild ? SPGIST_PLACEHOLDER : SPGIST_REDIRECT,
								SPGIST_PLACEHOLDER,
								fc_blknoDst,
								fc_toInsert[fc_nInsert - 1]);

		PageSetLSN(fc_page, fc_lsn);
		MarkBufferDirty(fc_buffer);
	}
	if (BufferIsValid(fc_buffer))
		UnlockReleaseBuffer(fc_buffer);

	/* 并更新父级下行链接 */
	if (XLogReadBufferForRedo(fc_record, 2, &fc_buffer) == BLK_NEEDS_REDO)
	{
		SpGistInnerTuple fc_tuple;

		fc_page = BufferGetPage(fc_buffer);

		fc_tuple = (SpGistInnerTuple) PageGetItem(fc_page,
											   PageGetItemId(fc_page, fc_xldata->offnumParent));

		spgUpdateNodeLink(fc_tuple, fc_xldata->nodeI,
						  fc_blknoDst, fc_toInsert[fc_nInsert - 1]);

		PageSetLSN(fc_page, fc_lsn);
		MarkBufferDirty(fc_buffer);
	}
	if (BufferIsValid(fc_buffer))
		UnlockReleaseBuffer(fc_buffer);
}

static void fc_spgRedoAddNode(XLogReaderState *fc_record)
{
	XLogRecPtr	fc_lsn = fc_record->EndRecPtr;
	char	   *fc_ptr = XLogRecGetData(fc_record);
	spgxlogAddNode *fc_xldata = (spgxlogAddNode *) fc_ptr;
	char	   *fc_innerTuple;
	SpGistInnerTupleData fc_innerTupleHdr;
	SpGistState fc_state;
	Buffer		fc_buffer;
	Page		fc_page;
	XLogRedoAction fc_action;

	fc_ptr += sizeof(spgxlogAddNode);
	fc_innerTuple = fc_ptr;
	/* 元组未对齐，因此需要复制以访问其头部 */
	memcpy(&fc_innerTupleHdr, fc_innerTuple, sizeof(SpGistInnerTupleData));

	fc_fillFakeState(&fc_state, fc_xldata->stateSrc);

	if (!XLogRecHasBlockRef(fc_record, 1))
	{
		/* 就地更新 */
		Assert(fc_xldata->parentBlk == -1);
		if (XLogReadBufferForRedo(fc_record, 0, &fc_buffer) == BLK_NEEDS_REDO)
		{
			fc_page = BufferGetPage(fc_buffer);

			PageIndexTupleDelete(fc_page, fc_xldata->offnum);
			if (PageAddItem(fc_page, (Item) fc_innerTuple, fc_innerTupleHdr.size,
							fc_xldata->offnum,
							false, false) != fc_xldata->offnum)
				elog(ERROR, "failed to add item of size %u to SPGiST index page",
					 fc_innerTupleHdr.size);

			PageSetLSN(fc_page, fc_lsn);
			MarkBufferDirty(fc_buffer);
		}
		if (BufferIsValid(fc_buffer))
			UnlockReleaseBuffer(fc_buffer);
	}
	else
	{
		BlockNumber fc_blkno;
		BlockNumber fc_blknoNew;

		XLogRecGetBlockTag(fc_record, 0, NULL, NULL, &fc_blkno);
		XLogRecGetBlockTag(fc_record, 1, NULL, NULL, &fc_blknoNew);

		/*
		 * 在正常操作中，我们会同时锁定所有三个页面（源、目标和父页面）；但是在 WAL 重放中，只要按正确顺序逐个更新它应该是安全的。
		 * 我们必须在用重定向元组替换旧元组之前插入新元组。
		 */

		/* 首先安装新元组，以确保重定向有效 */
		if (fc_xldata->newPage)
		{
			/* AddNode 不用于空页 */
			fc_buffer = XLogInitBufferForRedo(fc_record, 1);
			SpGistInitBuffer(fc_buffer, 0);
			fc_action = BLK_NEEDS_REDO;
		}
		else
			fc_action = XLogReadBufferForRedo(fc_record, 1, &fc_buffer);
		if (fc_action == BLK_NEEDS_REDO)
		{
			fc_page = BufferGetPage(fc_buffer);

			fc_addOrReplaceTuple(fc_page, (Item) fc_innerTuple,
							  fc_innerTupleHdr.size, fc_xldata->offnumNew);

			/*
			 * 如果父页面在此同一页中，请立即更新它。
			 */
			if (fc_xldata->parentBlk == 1)
			{
				SpGistInnerTuple fc_parentTuple;

				fc_parentTuple = (SpGistInnerTuple) PageGetItem(fc_page,
															 PageGetItemId(fc_page, fc_xldata->offnumParent));

				spgUpdateNodeLink(fc_parentTuple, fc_xldata->nodeI,
								  fc_blknoNew, fc_xldata->offnumNew);
			}
			PageSetLSN(fc_page, fc_lsn);
			MarkBufferDirty(fc_buffer);
		}
		if (BufferIsValid(fc_buffer))
			UnlockReleaseBuffer(fc_buffer);

		/* 删除旧元组，用重定向或占位元组替换它 */
		if (XLogReadBufferForRedo(fc_record, 0, &fc_buffer) == BLK_NEEDS_REDO)
		{
			SpGistDeadTuple fc_dt;

			fc_page = BufferGetPage(fc_buffer);

			if (fc_state.isBuild)
				fc_dt = spgFormDeadTuple(&fc_state, SPGIST_PLACEHOLDER,
									  InvalidBlockNumber,
									  InvalidOffsetNumber);
			else
				fc_dt = spgFormDeadTuple(&fc_state, SPGIST_REDIRECT,
									  fc_blknoNew,
									  fc_xldata->offnumNew);

			PageIndexTupleDelete(fc_page, fc_xldata->offnum);
			if (PageAddItem(fc_page, (Item) fc_dt, fc_dt->size,
							fc_xldata->offnum,
							false, false) != fc_xldata->offnum)
				elog(ERROR, "failed to add item of size %u to SPGiST index page",
					 fc_dt->size);

			if (fc_state.isBuild)
				SpGistPageGetOpaque(fc_page)->nPlaceholder++;
			else
				SpGistPageGetOpaque(fc_page)->nRedirection++;

			/*
			 * 如果父页面在此同一页中，请立即更新它。
			 */
			if (fc_xldata->parentBlk == 0)
			{
				SpGistInnerTuple fc_parentTuple;

				fc_parentTuple = (SpGistInnerTuple) PageGetItem(fc_page,
															 PageGetItemId(fc_page, fc_xldata->offnumParent));

				spgUpdateNodeLink(fc_parentTuple, fc_xldata->nodeI,
								  fc_blknoNew, fc_xldata->offnumNew);
			}
			PageSetLSN(fc_page, fc_lsn);
			MarkBufferDirty(fc_buffer);
		}
		if (BufferIsValid(fc_buffer))
			UnlockReleaseBuffer(fc_buffer);

		/*
		 * 更新父级下行链接（如果我们还没有作为源或目标页面更新的一部分完成它）。
		 */
		if (fc_xldata->parentBlk == 2)
		{
			if (XLogReadBufferForRedo(fc_record, 2, &fc_buffer) == BLK_NEEDS_REDO)
			{
				SpGistInnerTuple fc_parentTuple;

				fc_page = BufferGetPage(fc_buffer);

				fc_parentTuple = (SpGistInnerTuple) PageGetItem(fc_page,
															 PageGetItemId(fc_page, fc_xldata->offnumParent));

				spgUpdateNodeLink(fc_parentTuple, fc_xldata->nodeI,
								  fc_blknoNew, fc_xldata->offnumNew);

				PageSetLSN(fc_page, fc_lsn);
				MarkBufferDirty(fc_buffer);
			}
			if (BufferIsValid(fc_buffer))
				UnlockReleaseBuffer(fc_buffer);
		}
	}
}

static void fc_spgRedoSplitTuple(XLogReaderState *fc_record)
{
	XLogRecPtr	fc_lsn = fc_record->EndRecPtr;
	char	   *fc_ptr = XLogRecGetData(fc_record);
	spgxlogSplitTuple *fc_xldata = (spgxlogSplitTuple *) fc_ptr;
	char	   *fc_prefixTuple;
	SpGistInnerTupleData fc_prefixTupleHdr;
	char	   *fc_postfixTuple;
	SpGistInnerTupleData fc_postfixTupleHdr;
	Buffer		fc_buffer;
	Page		fc_page;
	XLogRedoAction fc_action;

	fc_ptr += sizeof(spgxlogSplitTuple);
	fc_prefixTuple = fc_ptr;
	/* 前缀元组未对齐，因此需要复制以访问其头部 */
	memcpy(&fc_prefixTupleHdr, fc_prefixTuple, sizeof(SpGistInnerTupleData));
	fc_ptr += fc_prefixTupleHdr.size;
	fc_postfixTuple = fc_ptr;
	/* 后缀元组也未对齐 */
	memcpy(&fc_postfixTupleHdr, fc_postfixTuple, sizeof(SpGistInnerTupleData));

	/*
	 * 在正常操作中，我们会同时锁定两个页面；但是在 WAL 重放中，只要按正确顺序逐个更新它应该是安全的。
	 */

	/* 首先插入后缀元组以避免悬空链接 */
	if (!fc_xldata->postfixBlkSame)
	{
		if (fc_xldata->newPage)
		{
			fc_buffer = XLogInitBufferForRedo(fc_record, 1);
			/* SplitTuple 不用于空页 */
			SpGistInitBuffer(fc_buffer, 0);
			fc_action = BLK_NEEDS_REDO;
		}
		else
			fc_action = XLogReadBufferForRedo(fc_record, 1, &fc_buffer);
		if (fc_action == BLK_NEEDS_REDO)
		{
			fc_page = BufferGetPage(fc_buffer);

			fc_addOrReplaceTuple(fc_page, (Item) fc_postfixTuple,
							  fc_postfixTupleHdr.size, fc_xldata->offnumPostfix);

			PageSetLSN(fc_page, fc_lsn);
			MarkBufferDirty(fc_buffer);
		}
		if (BufferIsValid(fc_buffer))
			UnlockReleaseBuffer(fc_buffer);
	}

	/* 现在处理原始页面 */
	if (XLogReadBufferForRedo(fc_record, 0, &fc_buffer) == BLK_NEEDS_REDO)
	{
		fc_page = BufferGetPage(fc_buffer);

		PageIndexTupleDelete(fc_page, fc_xldata->offnumPrefix);
		if (PageAddItem(fc_page, (Item) fc_prefixTuple, fc_prefixTupleHdr.size,
						fc_xldata->offnumPrefix, false, false) != fc_xldata->offnumPrefix)
			elog(ERROR, "failed to add item of size %u to SPGiST index page",
				 fc_prefixTupleHdr.size);

		if (fc_xldata->postfixBlkSame)
			fc_addOrReplaceTuple(fc_page, (Item) fc_postfixTuple,
							  fc_postfixTupleHdr.size,
							  fc_xldata->offnumPostfix);

		PageSetLSN(fc_page, fc_lsn);
		MarkBufferDirty(fc_buffer);
	}
	if (BufferIsValid(fc_buffer))
		UnlockReleaseBuffer(fc_buffer);
}

static void fc_spgRedoPickSplit(XLogReaderState *fc_record)
{
	XLogRecPtr	fc_lsn = fc_record->EndRecPtr;
	char	   *fc_ptr = XLogRecGetData(fc_record);
	spgxlogPickSplit *fc_xldata = (spgxlogPickSplit *) fc_ptr;
	char	   *fc_innerTuple;
	SpGistInnerTupleData fc_innerTupleHdr;
	SpGistState fc_state;
	OffsetNumber *fc_toDelete;
	OffsetNumber *fc_toInsert;
	uint8	   *fc_leafPageSelect;
	Buffer		fc_srcBuffer;
	Buffer		fc_destBuffer;
	Buffer		fc_innerBuffer;
	Page		fc_srcPage;
	Page		fc_destPage;
	Page		fc_page;
	int			fc_i;
	BlockNumber fc_blknoInner;
	XLogRedoAction fc_action;

	XLogRecGetBlockTag(fc_record, 2, NULL, NULL, &fc_blknoInner);

	fc_fillFakeState(&fc_state, fc_xldata->stateSrc);

	fc_ptr += SizeOfSpgxlogPickSplit;
	fc_toDelete = (OffsetNumber *) fc_ptr;
	fc_ptr += sizeof(OffsetNumber) * fc_xldata->nDelete;
	fc_toInsert = (OffsetNumber *) fc_ptr;
	fc_ptr += sizeof(OffsetNumber) * fc_xldata->nInsert;
	fc_leafPageSelect = (uint8 *) fc_ptr;
	fc_ptr += sizeof(uint8) * fc_xldata->nInsert;

	fc_innerTuple = fc_ptr;
	/* 内部元组未对齐，因此需要复制以访问其头部 */
	memcpy(&fc_innerTupleHdr, fc_innerTuple, sizeof(SpGistInnerTupleData));
	fc_ptr += fc_innerTupleHdr.size;

	/* 现在指针指向叶子元组的列表 */

	if (fc_xldata->isRootSplit)
	{
		/* 当分割根节点时，我们只在新内部的伪装下接触它 */
		fc_srcBuffer = InvalidBuffer;
		fc_srcPage = NULL;
	}
	else if (fc_xldata->initSrc)
	{
		/* 重新初始化源页面 */
		fc_srcBuffer = XLogInitBufferForRedo(fc_record, 0);
		fc_srcPage = (Page) BufferGetPage(fc_srcBuffer);

		SpGistInitBuffer(fc_srcBuffer,
						 SPGIST_LEAF | (fc_xldata->storesNulls ? SPGIST_NULLS : 0));
		/* 在完成之前不要更新 LSN 等 */
	}
	else
	{
		/*
		 * 从源页面删除指定的元组。(如果我们在热备用状态下，我们
		 * 需要在完成插入叶子元组和新的内部元组之前保持对页面的锁定，
		 * 否则添加的重定向元组将成为一个悬空的链接。)
		 */
		fc_srcPage = NULL;
		if (XLogReadBufferForRedo(fc_record, 0, &fc_srcBuffer) == BLK_NEEDS_REDO)
		{
			fc_srcPage = BufferGetPage(fc_srcBuffer);

			/*
			 * 在这里我们比 doPickSplit() 轻松一些，因为我们
			 * 已经知道内部元组的位置，所以现在可以注入
			 * 正确的重定向元组。
			 */
			if (!fc_state.isBuild)
				spgPageIndexMultiDelete(&fc_state, fc_srcPage,
										fc_toDelete, fc_xldata->nDelete,
										SPGIST_REDIRECT,
										SPGIST_PLACEHOLDER,
										fc_blknoInner,
										fc_xldata->offnumInner);
			else
				spgPageIndexMultiDelete(&fc_state, fc_srcPage,
										fc_toDelete, fc_xldata->nDelete,
										SPGIST_PLACEHOLDER,
										SPGIST_PLACEHOLDER,
										InvalidBlockNumber,
										InvalidOffsetNumber);

			/* 在完成之前不要更新 LSN 等 */
		}
	}

	/* 如果有的话，尝试访问目标页面 */
	if (!XLogRecHasBlockRef(fc_record, 1))
	{
		fc_destBuffer = InvalidBuffer;
		fc_destPage = NULL;
	}
	else if (fc_xldata->initDest)
	{
		/* 重新初始化目标页面 */
		fc_destBuffer = XLogInitBufferForRedo(fc_record, 1);
		fc_destPage = (Page) BufferGetPage(fc_destBuffer);

		SpGistInitBuffer(fc_destBuffer,
						 SPGIST_LEAF | (fc_xldata->storesNulls ? SPGIST_NULLS : 0));
		/* 在完成之前不要更新 LSN 等 */
	}
	else
	{
		/*
		 * 在完整页面图像的情况下，我们可能可以立即释放页面锁，
		 * 但为了安全起见，我们还是等到稍后再释放。
		 */
		if (XLogReadBufferForRedo(fc_record, 1, &fc_destBuffer) == BLK_NEEDS_REDO)
			fc_destPage = (Page) BufferGetPage(fc_destBuffer);
		else
			fc_destPage = NULL;	/* 不要进行任何页面更新 */
	}

	/* 将叶子元组恢复到源页面和/或目标页面 */
	for (fc_i = 0; fc_i < fc_xldata->nInsert; fc_i++)
	{
		char	   *fc_leafTuple;
		SpGistLeafTupleData fc_leafTupleHdr;

		/* 这些元组没有对齐，因此必须复制以访问大小字段。 */
		fc_leafTuple = fc_ptr;
		memcpy(&fc_leafTupleHdr, fc_leafTuple, sizeof(SpGistLeafTupleData));
		fc_ptr += fc_leafTupleHdr.size;

		fc_page = fc_leafPageSelect[fc_i] ? fc_destPage : fc_srcPage;
		if (fc_page == NULL)
			continue;			/* 不需要触碰这个页面 */

		fc_addOrReplaceTuple(fc_page, (Item) fc_leafTuple, fc_leafTupleHdr.size,
						  fc_toInsert[fc_i]);
	}

	/* 如果需要，更新源页面和目标页面的 LSN */
	if (fc_srcPage != NULL)
	{
		PageSetLSN(fc_srcPage, fc_lsn);
		MarkBufferDirty(fc_srcBuffer);
	}
	if (fc_destPage != NULL)
	{
		PageSetLSN(fc_destPage, fc_lsn);
		MarkBufferDirty(fc_destBuffer);
	}

	/* 恢复新的内部元组 */
	if (fc_xldata->initInner)
	{
		fc_innerBuffer = XLogInitBufferForRedo(fc_record, 2);
		SpGistInitBuffer(fc_innerBuffer, (fc_xldata->storesNulls ? SPGIST_NULLS : 0));
		fc_action = BLK_NEEDS_REDO;
	}
	else
		fc_action = XLogReadBufferForRedo(fc_record, 2, &fc_innerBuffer);

	if (fc_action == BLK_NEEDS_REDO)
	{
		fc_page = BufferGetPage(fc_innerBuffer);

		fc_addOrReplaceTuple(fc_page, (Item) fc_innerTuple, fc_innerTupleHdr.size,
						  fc_xldata->offnumInner);

		/* 如果内部元组也是父节点，则在这里更新链接 */
		if (fc_xldata->innerIsParent)
		{
			SpGistInnerTuple fc_parent;

			fc_parent = (SpGistInnerTuple) PageGetItem(fc_page,
													PageGetItemId(fc_page, fc_xldata->offnumParent));
			spgUpdateNodeLink(fc_parent, fc_xldata->nodeI,
							  fc_blknoInner, fc_xldata->offnumInner);
		}

		PageSetLSN(fc_page, fc_lsn);
		MarkBufferDirty(fc_innerBuffer);
	}
	if (BufferIsValid(fc_innerBuffer))
		UnlockReleaseBuffer(fc_innerBuffer);

	/*
	 * 现在我们可以释放叶子页面的锁定。在更新父节点下链之前这样做是可以的。
	 */
	if (BufferIsValid(fc_srcBuffer))
		UnlockReleaseBuffer(fc_srcBuffer);
	if (BufferIsValid(fc_destBuffer))
		UnlockReleaseBuffer(fc_destBuffer);

	/* 更新父节点下链，除非我们在上述操作中已完成 */
	if (XLogRecHasBlockRef(fc_record, 3))
	{
		Buffer		fc_parentBuffer;

		if (XLogReadBufferForRedo(fc_record, 3, &fc_parentBuffer) == BLK_NEEDS_REDO)
		{
			SpGistInnerTuple fc_parent;

			fc_page = BufferGetPage(fc_parentBuffer);

			fc_parent = (SpGistInnerTuple) PageGetItem(fc_page,
													PageGetItemId(fc_page, fc_xldata->offnumParent));
			spgUpdateNodeLink(fc_parent, fc_xldata->nodeI,
							  fc_blknoInner, fc_xldata->offnumInner);

			PageSetLSN(fc_page, fc_lsn);
			MarkBufferDirty(fc_parentBuffer);
		}
		if (BufferIsValid(fc_parentBuffer))
			UnlockReleaseBuffer(fc_parentBuffer);
	}
	else
		Assert(fc_xldata->innerIsParent || fc_xldata->isRootSplit);
}

static void fc_spgRedoVacuumLeaf(XLogReaderState *fc_record)
{
	XLogRecPtr	fc_lsn = fc_record->EndRecPtr;
	char	   *fc_ptr = XLogRecGetData(fc_record);
	spgxlogVacuumLeaf *fc_xldata = (spgxlogVacuumLeaf *) fc_ptr;
	OffsetNumber *fc_toDead;
	OffsetNumber *fc_toPlaceholder;
	OffsetNumber *fc_moveSrc;
	OffsetNumber *fc_moveDest;
	OffsetNumber *fc_chainSrc;
	OffsetNumber *fc_chainDest;
	SpGistState fc_state;
	Buffer		fc_buffer;
	Page		fc_page;
	int			fc_i;

	fc_fillFakeState(&fc_state, fc_xldata->stateSrc);

	fc_ptr += SizeOfSpgxlogVacuumLeaf;
	fc_toDead = (OffsetNumber *) fc_ptr;
	fc_ptr += sizeof(OffsetNumber) * fc_xldata->nDead;
	fc_toPlaceholder = (OffsetNumber *) fc_ptr;
	fc_ptr += sizeof(OffsetNumber) * fc_xldata->nPlaceholder;
	fc_moveSrc = (OffsetNumber *) fc_ptr;
	fc_ptr += sizeof(OffsetNumber) * fc_xldata->nMove;
	fc_moveDest = (OffsetNumber *) fc_ptr;
	fc_ptr += sizeof(OffsetNumber) * fc_xldata->nMove;
	fc_chainSrc = (OffsetNumber *) fc_ptr;
	fc_ptr += sizeof(OffsetNumber) * fc_xldata->nChain;
	fc_chainDest = (OffsetNumber *) fc_ptr;

	if (XLogReadBufferForRedo(fc_record, 0, &fc_buffer) == BLK_NEEDS_REDO)
	{
		fc_page = BufferGetPage(fc_buffer);

		spgPageIndexMultiDelete(&fc_state, fc_page,
								fc_toDead, fc_xldata->nDead,
								SPGIST_DEAD, SPGIST_DEAD,
								InvalidBlockNumber,
								InvalidOffsetNumber);

		spgPageIndexMultiDelete(&fc_state, fc_page,
								fc_toPlaceholder, fc_xldata->nPlaceholder,
								SPGIST_PLACEHOLDER, SPGIST_PLACEHOLDER,
								InvalidBlockNumber,
								InvalidOffsetNumber);

		/* 请参见 vacuumLeafPage() 中的注释 */
		for (fc_i = 0; fc_i < fc_xldata->nMove; fc_i++)
		{
			ItemId		fc_idSrc = PageGetItemId(fc_page, fc_moveSrc[fc_i]);
			ItemId		fc_idDest = PageGetItemId(fc_page, fc_moveDest[fc_i]);
			ItemIdData	fc_tmp;

			fc_tmp = *fc_idSrc;
			*fc_idSrc = *fc_idDest;
			*fc_idDest = fc_tmp;
		}

		spgPageIndexMultiDelete(&fc_state, fc_page,
								fc_moveSrc, fc_xldata->nMove,
								SPGIST_PLACEHOLDER, SPGIST_PLACEHOLDER,
								InvalidBlockNumber,
								InvalidOffsetNumber);

		for (fc_i = 0; fc_i < fc_xldata->nChain; fc_i++)
		{
			SpGistLeafTuple fc_lt;

			fc_lt = (SpGistLeafTuple) PageGetItem(fc_page,
											   PageGetItemId(fc_page, fc_chainSrc[fc_i]));
			Assert(fc_lt->tupstate == SPGIST_LIVE);
			SGLT_SET_NEXTOFFSET(fc_lt, fc_chainDest[fc_i]);
		}

		PageSetLSN(fc_page, fc_lsn);
		MarkBufferDirty(fc_buffer);
	}
	if (BufferIsValid(fc_buffer))
		UnlockReleaseBuffer(fc_buffer);
}

static void fc_spgRedoVacuumRoot(XLogReaderState *fc_record)
{
	XLogRecPtr	fc_lsn = fc_record->EndRecPtr;
	char	   *fc_ptr = XLogRecGetData(fc_record);
	spgxlogVacuumRoot *fc_xldata = (spgxlogVacuumRoot *) fc_ptr;
	OffsetNumber *fc_toDelete;
	Buffer		fc_buffer;
	Page		fc_page;

	fc_toDelete = fc_xldata->offsets;

	if (XLogReadBufferForRedo(fc_record, 0, &fc_buffer) == BLK_NEEDS_REDO)
	{
		fc_page = BufferGetPage(fc_buffer);

		/* 元组编号是按顺序排列的 */
		PageIndexMultiDelete(fc_page, fc_toDelete, fc_xldata->nDelete);

		PageSetLSN(fc_page, fc_lsn);
		MarkBufferDirty(fc_buffer);
	}
	if (BufferIsValid(fc_buffer))
		UnlockReleaseBuffer(fc_buffer);
}

static void fc_spgRedoVacuumRedirect(XLogReaderState *fc_record)
{
	XLogRecPtr	fc_lsn = fc_record->EndRecPtr;
	char	   *fc_ptr = XLogRecGetData(fc_record);
	spgxlogVacuumRedirect *fc_xldata = (spgxlogVacuumRedirect *) fc_ptr;
	OffsetNumber *fc_itemToPlaceholder;
	Buffer		fc_buffer;

	fc_itemToPlaceholder = fc_xldata->offsets;

	/*
	 * 如果正在移除任何重定向元组，确保没有
	 * 活动的热备用事务可能需要看到它们。
	 */
	if (InHotStandby)
	{
		if (TransactionIdIsValid(fc_xldata->newestRedirectXid))
		{
			RelFileNode fc_node;

			XLogRecGetBlockTag(fc_record, 0, &fc_node, NULL, NULL);
			ResolveRecoveryConflictWithSnapshot(fc_xldata->newestRedirectXid,
												fc_node);
		}
	}

	if (XLogReadBufferForRedo(fc_record, 0, &fc_buffer) == BLK_NEEDS_REDO)
	{
		Page		fc_page = BufferGetPage(fc_buffer);
		SpGistPageOpaque fc_opaque = SpGistPageGetOpaque(fc_page);
		int			fc_i;

		/* 将重定向指针转换为普通占位符 */
		for (fc_i = 0; fc_i < fc_xldata->nToPlaceholder; fc_i++)
		{
			SpGistDeadTuple fc_dt;

			fc_dt = (SpGistDeadTuple) PageGetItem(fc_page,
											   PageGetItemId(fc_page, fc_itemToPlaceholder[fc_i]));
			Assert(fc_dt->tupstate == SPGIST_REDIRECT);
			fc_dt->tupstate = SPGIST_PLACEHOLDER;
			ItemPointerSetInvalid(&fc_dt->pointer);
		}

		Assert(fc_opaque->nRedirection >= fc_xldata->nToPlaceholder);
		fc_opaque->nRedirection -= fc_xldata->nToPlaceholder;
		fc_opaque->nPlaceholder += fc_xldata->nToPlaceholder;

		/* 移除页面末尾的占位符元组 */
		if (fc_xldata->firstPlaceholder != InvalidOffsetNumber)
		{
			int			fc_max = PageGetMaxOffsetNumber(fc_page);
			OffsetNumber *fc_toDelete;

			fc_toDelete = palloc(sizeof(OffsetNumber) * fc_max);

			for (fc_i = fc_xldata->firstPlaceholder; fc_i <= fc_max; fc_i++)
				fc_toDelete[fc_i - fc_xldata->firstPlaceholder] = fc_i;

			fc_i = fc_max - fc_xldata->firstPlaceholder + 1;
			Assert(fc_opaque->nPlaceholder >= fc_i);
			fc_opaque->nPlaceholder -= fc_i;

			/* 数组已排序，因此可以使用 PageIndexMultiDelete */
			PageIndexMultiDelete(fc_page, fc_toDelete, fc_i);

			pfree(fc_toDelete);
		}

		PageSetLSN(fc_page, fc_lsn);
		MarkBufferDirty(fc_buffer);
	}
	if (BufferIsValid(fc_buffer))
		UnlockReleaseBuffer(fc_buffer);
}

void spg_redo(XLogReaderState *fc_record)
{
	uint8		fc_info = XLogRecGetInfo(fc_record) & ~XLR_INFO_MASK;
	MemoryContext fc_oldCxt;

	fc_oldCxt = MemoryContextSwitchTo(opCtx);
	switch (fc_info)
	{
		case XLOG_SPGIST_ADD_LEAF:
			fc_spgRedoAddLeaf(fc_record);
			break;
		case XLOG_SPGIST_MOVE_LEAFS:
			fc_spgRedoMoveLeafs(fc_record);
			break;
		case XLOG_SPGIST_ADD_NODE:
			fc_spgRedoAddNode(fc_record);
			break;
		case XLOG_SPGIST_SPLIT_TUPLE:
			fc_spgRedoSplitTuple(fc_record);
			break;
		case XLOG_SPGIST_PICKSPLIT:
			fc_spgRedoPickSplit(fc_record);
			break;
		case XLOG_SPGIST_VACUUM_LEAF:
			fc_spgRedoVacuumLeaf(fc_record);
			break;
		case XLOG_SPGIST_VACUUM_ROOT:
			fc_spgRedoVacuumRoot(fc_record);
			break;
		case XLOG_SPGIST_VACUUM_REDIRECT:
			fc_spgRedoVacuumRedirect(fc_record);
			break;
		default:
			elog(PANIC, "spg_redo: unknown op code %u", fc_info);
	}

	MemoryContextSwitchTo(fc_oldCxt);
	MemoryContextReset(opCtx);
}

void spg_xlog_startup(void)
{
	opCtx = AllocSetContextCreate(CurrentMemoryContext,
								  "SP-GiST temporary context",
								  ALLOCSET_DEFAULT_SIZES);
}

void spg_xlog_cleanup(void)
{
	MemoryContextDelete(opCtx);
	opCtx = NULL;
}

/*
 * 在对 SpGist 页面执行一致性检查之前对其进行掩码处理。
 */
void spg_mask(char *fc_pagedata, BlockNumber fc_blkno)
{
	Page		fc_page = (Page) fc_pagedata;
	PageHeader	fc_pagehdr = (PageHeader) fc_page;

	mask_page_lsn_and_checksum(fc_page);

	mask_page_hint_bits(fc_page);

	/*
	 * 掩码未使用的空间，但只有在页面的 pd_lower 看起来已
	 * 正确设置的情况下。
	 */
	if (fc_pagehdr->pd_lower >= SizeOfPageHeaderData)
		mask_unused_space(fc_page);
}
